package me.sdau.spider.kafkaprocess.kafka.nlp;

import com.alibaba.fastjson.JSONObject;
import me.sdau.spider.kafkaprocess.kafka.dto.ConfigDTO;
import me.sdau.spider.kafkaprocess.kafka.nlp.sdk.NlpFactory;
import me.sdau.spider.kafkaprocess.kafka.nlp.sdk.NlpSdk;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;

import java.util.Properties;


/**
 * @author zj
 * @date 2018/4/2 11:15
 */
public class NLPBuilder {

    private static Properties props = new Properties();

    static {
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-counts");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "172.16.18.6:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    }

    public static KafkaStreams getStreams(ConfigDTO config) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> stream = builder.stream(config.getTopic());

//        StampedLock lock = new StampedLock();

        NlpSdk.Arg arg = new NlpSdk.Arg();
        stream.foreach((key, value) -> {
            JSONObject json = JSONObject.parseObject(value);
            arg.setJsonObject(json);
            arg.setContent(json.getString("content"));
            arg.setTitle(json.getString("title"));
        });

        for (String processor : config.getProcessors()) {
            stream = stream.map((key, value) -> {
                JSONObject json = getJsonObject(arg, processor);

                return KeyValue.pair(key, json.toString());
            });
        }
        stream.to("testII");

        props.put(StreamsConfig.APPLICATION_ID_CONFIG, config.getTopic());
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
        return streams;
    }

    private static JSONObject getJsonObject(NlpSdk.Arg arg, String processor) {
        String nlpMethod = NlpFactory.get(processor)
                .get(arg.getTitle(), arg.getContent());
        JSONObject json = arg.getJsonObject();
        json.put(processor, nlpMethod);
        return json;
    }

}
